package com.spark.WordCount;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

public class Spark_WordCount_Java {
    public static void main(String[] args) {
        //1、建立和Spark框架的连接 创建SparkContext
        //jdbc：connection
        SparkConf sparkConf=new SparkConf()
                .setAppName("WordCount")
        .setMaster("local");
        JavaSparkContext sparkContext=new JavaSparkContext(sparkConf);
        //2、执行业务操作
        //2.1：读取文件 加载数据，读取一行数据
        String path="datas/1.txt";
        if(args.length==1){
            path=args[0];
        }
        JavaRDD<String> linesRDD=sparkContext.textFile(path);
        //2.2：单词拆分（将一行数据拆分为多个单词）
        JavaRDD<String> wordsRDD=linesRDD.flatMap(new FlatMapFunction<String, String>() {

            @Override
            public Iterator<String> call(String line) throws Exception {
                return Arrays.asList(line.split(" ")).iterator();
            }
        });

        //2.3：迭代words，将words中的每一个word转化为（word，1）
        JavaPairRDD<String,Integer> pairRDD=wordsRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<String ,Integer>(word,1);
            }
        });

        //2.4：根据Key(word)进行分组聚合
        JavaPairRDD<String,Integer> wordCountRDD=pairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer1, Integer integer2) throws Exception {
                return integer1+integer2;
            }
        });
        //2.5：输出统计结果
        wordCountRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> tuple) throws Exception {
                System.out.println(tuple._1+"--"+tuple._2);
            }
        });

        //3、关闭连接,停止SparkContext
        sparkContext.stop();
    }
}

